import logging
import queue
import threading
import time
import atexit
from concurrent.futures import ThreadPoolExecutor

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)


class MyThreadPoolExecutor(ThreadPoolExecutor):
    def __init__(self, max_workers=None, thread_name_prefix='', qsize=0):
        '''
        max_workers: 最多同时有多少个线程在执行任务,其余的任务会排队等着。
        qsize: 线程池内部的“等待队列”的最大容量（可以缓存多少个还没执行的任务）

        举例：
        max_workers = 4：有 4 个线程可以并行处理任务
        qsize = 100：排队等着执行的任务最多有 100 个
        如果你连续 submit() 了 200 个任务：
            前 4 个被线程立刻处理
            接下来的 100 个进了队列
            剩下的 96 个会怎样？→ 被阻塞或抛出异常，取决于你是否对 queue.Queue 做了处理（put_nowait()、put(timeout) 等）
        '''

        super().__init__(max_workers=max_workers, thread_name_prefix=thread_name_prefix)
        self.max_qsize = qsize

        # 默认ThreadPoolExecutor的队列是无限长！这里加了一个限制！
        self._work_queue = queue.Queue(qsize) # type: ignore

        # 添加队列监控线程 self.log_queue_size（便于观察是否线程池处理能力不足、判断是否需要扩容线程池或优化任务调度）
        threading.Thread(target=self.log_queue_size, daemon=True).start()

        # 程序要关闭时 确保线程池中的所有任务都执行完
        atexit.register(self.shutdown_gracefully)

    def log_queue_size(self):
        while True:
            logger.info(f'[ThreadPool] 当前等待队列长度: {self.get_qsize()}')
            time.sleep(5)

    def get_qsize(self):
        return self._work_queue.qsize()

    def shutdown_gracefully(self, wait=True):
        logger.info("[ThreadPool] 正在优雅关闭线程池...")
        super().shutdown(wait=wait)

        # TODO 模拟线程池中还有任务在执行..............
        time.sleep(1.5)

        logger.info("[ThreadPool] 线程池关闭完成。")


# 全局线程池对象
general_business_executor: MyThreadPoolExecutor = None

__pool_lock = threading.Lock()


# Notice 必须先 init_thread_pool 再 get_thread_pool，否则 get_thread_pool().submit(...) 会抛 NontType 错误!
def init_thread_pool(max_workers=4, qsize=100):
    # Notice 线程安全的惰性单例模式: 防止多个模块同时调用产生的竞争条件问题
    global general_business_executor
    if general_business_executor is None:
        with __pool_lock:
            # 双重检查
            if general_business_executor is None:
                general_business_executor = MyThreadPoolExecutor(
                    max_workers=max_workers, thread_name_prefix="general_business", qsize=qsize
                )

    return general_business_executor


def get_thread_pool():
    return general_business_executor


# -----------------------------------------------------------------------------------
class NoneResult:
    def result(self):
        return None


def after_call_error_handler(e: Exception):
    e_str = str(e)
    logger.error('get exception %s', e_str, exc_info=True, stack_info=False)

    # TODO 发邮件...
    # send_email(e_str, '', '')

    return e_str


# Notice 异步任务装饰器
# 将同步函数封装为异步任务，并统一交由线程池 general_business_executor 执行
def async_thread(func):
    # 获取参数
    def thread_wrapper(*args, **kwargs):

        # Notice 线程池中的线程具体执行的代码！
        def real_func_wrapper(*args, **kwargs):
            try:
                func_res = func(*args, **kwargs)
                return func_res
            except Exception as e:
                after_call_error_handler(e)

        # 等待队列满了，根据调用者意愿决定：
        # （1）抛出异常（默认行为 raise_error=True），表示当前系统繁忙，无法再接受新的异步任务了
        # （2）静默忽略这个任务，仅返回一个伪造的空 Future（通过 NoneResult() 实现）
        # Notice "被装饰的函数" 需要设置 raise_error=True 或者 raise_error=False
        if general_business_executor.max_qsize != 0 and general_business_executor.get_qsize() == general_business_executor.max_qsize:
            if kwargs.get('raise_error', True):
                e_str = f'execute reach max qsize, get_qsize={general_business_executor.get_qsize()}, args: f{args}, f{kwargs}'
                logger.error('get exception %s', e_str, exc_info=True, stack_info=False)
                raise e_str
            else:
                return NoneResult()

        future = general_business_executor.submit(real_func_wrapper, *args, **kwargs)

        return future

    return thread_wrapper
